package org.yonggan.flume;

import com.google.common.collect.Lists;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;
import java.util.List;

public class LoggerFilterInterceptor implements Interceptor {

    private static long count;

    private static final Logger LOGGER = LoggerFactory.getLogger(LoggerFilterInterceptor.class);

    private String DEFUALT_CHARSET = "utf-8";

    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {

        try {
            if (event == null) return null;
            // 将数据 转换成为 json 实现过滤
            String json = new String(event.getBody(), DEFUALT_CHARSET);
            // 实现数据过滤
            if (isNeedCollect(json)) {
                return event;
            }
            count ++;
        } catch (Exception e) {
            if (LOGGER.isErrorEnabled()) LOGGER.error("日志解析异常 原因：${}", e.getMessage());
        }
        return null;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        List<Event> intercepted = Lists.newArrayListWithCapacity(list.size());
        for (Event event : list) {
            Event interceptedEvent = intercept(event);
            if (interceptedEvent != null) {
                intercepted.add(interceptedEvent);
            }
        }
        return intercepted;
    }

    @Override
    public void close() {

    }


    /**
     * 过滤 判断 是否符合
     *
     * @return
     */
    private Boolean isNeedCollect(String json) {
        return json.contains("reChargeNotifyReq");
    }

    /**
     *
     */
    public static class Builder implements Interceptor.Builder {

        //使用Builder初始化Interceptor
        public Interceptor build() {
            return new LoggerFilterInterceptor();
        }

        public void configure(Context context) {

        }
    }
}
